package com.atguigu.day03;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Integer>> stream = env
                .fromElements(
                        "hello world",
                        "hello world"
                )
                .flatMap(
                        (String in, Collector<Tuple2<String, Integer>> out) -> {
                            String[] arr = in.split(" ");
                            for (String word : arr) out.collect(Tuple2.of(word, 1));
                        }
                )
                .returns(Types.TUPLE(Types.STRING, Types.INT));

        stream
                .keyBy(r -> r.f0)
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    // 输入数据和累加器的累加规则
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                        return Tuple2.of(
                                value1.f0,
                                value1.f1 + value2.f1
                        );
                    }
                })
                .print();

        stream
                .keyBy(r -> r.f0)
                .reduce((r1, r2) -> Tuple2.of(r1.f0, r1.f1 + r2.f1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .print();

        env.execute();
    }
}
